package com.atguigu.flink.chapter06.watermark;

import com.atguigu.flink.bean.WaterSensor;
import com.atguigu.flink.chapter05.function.WaterSensorMapFunction;
import com.atguigu.flink.util.AtguiguUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;

public class WaterMarkDemo4 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        SingleOutputStreamOperator<String> mainStream = env.socketTextStream("hadoop102", 8888)
                .map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))    //设置最大乱序程度
                                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                                    @Override
                                    public long extractTimestamp(WaterSensor elements,   //流中的元素
                                                                 long l) {
                                        return elements.getTs();        //是一个毫秒值
                                    }
                                })
                                //当一个并行度的水印超过30s不更新，就其他的为准
                                .withIdleness(Duration.ofSeconds(30))
                )
                .keyBy(WaterSensor::getId)
                //定义窗口：基于事件时间的滚动窗口，窗口的长度是5s
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .allowedLateness(Time.seconds(2))    //允许迟到的数据
                .sideOutputLateData(new OutputTag<WaterSensor>("late") {}) //迟到的数据输出
                //窗口处理函数
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {
                        List<WaterSensor> list = AtguiguUtil.toList(elements);
                        out.collect(ctx.window().getStart() + " " + ctx.window().getEnd() + " " + list);
                    }
                });
        mainStream.print("main");
        mainStream.getSideOutput(new OutputTag<WaterSensor>("late"){}).print("late");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}



//public class WaterMarkDemo4 {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//
//        SingleOutputStreamOperator<String> mainStream = env.socketTextStream("hadoop102", 8888)
//                .map(new WaterSensorMapFunction())
//                .assignTimestampsAndWatermarks(
//                        WatermarkStrategy
//                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))    //设置最大乱序程度
//                                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
//                                    @Override
//                                    public long extractTimestamp(WaterSensor elements, //流中的元素
//                                                                 long l) {
//                                        return elements.getTs();   //是一个毫秒值
//                                    }
//                                })
//                                //当一个并行度的水印超过30s不更新，就以其他为准
//                                .withIdleness(Duration.ofSeconds(30))
//                )
//                .keyBy(WaterSensor::getId)
//                //定义窗口：基于事件时间的滚动窗口，窗口的长度为5s
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .allowedLateness(Time.seconds(3))    //允许延迟的数据
//                .sideOutputLateData(new OutputTag<WaterSensor>("late") {
//                })  //延迟的侧输出
//                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
//                    @Override
//                    public void process(String key,
//                                        Context ctx,
//                                        Iterable<WaterSensor> elements,
//                                        Collector<String> out) throws Exception {
//                        List<WaterSensor> list = AtguiguUtil.toList(elements);
//                        out.collect(ctx.window().getStart() + " " + ctx.window().getEnd() + " " + list);
//                    }
//                });
//        mainStream.print("late");
//        mainStream.getSideOutput(new OutputTag<WaterSensor>("late"){}).print("late");
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}


//public class WaterMarkDemo4 {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//
//        SingleOutputStreamOperator<String> mainStream = env.socketTextStream("hadoop102", 8888)
//                .map(new WaterSensorMapFunction())
//                .assignTimestampsAndWatermarks(
//                        WatermarkStrategy
//                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
//                                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
//                                    @Override
//                                    public long extractTimestamp(WaterSensor elements,
//                                                                 long l) {
//                                        return elements.getTs();
//                                    }
//                                })
//                                //当一个并行度的水印超过30s不更新，就以其他为准
//                                .withIdleness(Duration.ofSeconds(2))
//                )
//                .keyBy(WaterSensor::getId)
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .allowedLateness(Time.seconds(3)) //允许延迟的数据
//                .sideOutputLateData(new OutputTag<WaterSensor>("late") {
//                }) //延迟的数据
//                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
//                    @Override
//                    public void process(String key,
//                                        Context ctx,
//                                        Iterable<WaterSensor> elements,
//                                        Collector<String> out) throws Exception {
//                        List<WaterSensor> list = AtguiguUtil.toList(elements);
//                        out.collect(ctx.window().getStart() + " " + ctx.window().getEnd() + " " + list);
//                    }
//                });
//        mainStream.print("mian");
//
//        mainStream.getSideOutput(new OutputTag<WaterSensor>("late"){}).print("late");
//
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}